package com.heima.article.stream;

import cn.hutool.json.JSONUtil;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.article.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * @author TheCai
 * @date 2021/6/10 20:02
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {
  @Override
  public String listenerTopic() { // 流监听主题
    return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
  }

  @Override
  public String sendTopic() { // 流处理完毕  发送到的主题
    return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
  }

  @Override
  public KStream<String, String> getService(KStream<String, String> stream) {
    return stream
        .map(
            (key, value) -> {
              log.info("流式处理器被触发  ,  接收到的消息内容为==============>>" + value);
              UpdateArticleMess updateArticleMess = JSONUtil.toBean(value, UpdateArticleMess.class);
              String newKey =
                  updateArticleMess.getArticleId()
                      + ":"
                      + updateArticleMess.getType().name(); // 文章id:文章类型
              String newValue = updateArticleMess.getAdd().toString();

              return new KeyValue<>(newKey, newValue); // key:1241:LIKES   value:1
            })
        .groupByKey() // 按照key进行分支
        .windowedBy(TimeWindows.of(Duration.ofSeconds(10))) // 时间窗口  每隔多久统计一次
        .aggregate(
            () -> "0",
            (k, v, perR) -> { // 定义分组后的聚合处理方式
              // k:每条数据的key  v:每条数据value  preR: 上一次啊计算的结果
              return String.valueOf(Integer.valueOf(perR) + Integer.valueOf(v));
            })
        .toStream()
        .map(
            (key, value) -> { // key: 对象(原来分组的key   时间戳) value:聚合计算的结果
              String articleIdAndType = key.key();
              Integer result = Integer.valueOf(value); // value :  聚合计算结果
              ArticleVisitStreamMess streamMess = formatObj(articleIdAndType, result); // 格式化对象
              return new KeyValue<>("", JSONUtil.toJsonStr(streamMess));
            }) // 格式化key: value
    ;
  }

  /**
   * 封装对象
   *
   * @param articleIdAndType 文章 id:类型
   * @param result 统计的数量
   * @return
   */
  private ArticleVisitStreamMess formatObj(String articleIdAndType, Integer result) {
    ArticleVisitStreamMess streamMess = new ArticleVisitStreamMess();
    String[] split = articleIdAndType.split(":"); // [文章id,;类型]
    String articleId = split[0];
    String articleTypeName = split[1];
    streamMess.setArticleId(Long.valueOf(articleId)); // 设置文章id

    if (articleTypeName.equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())) { // 阅读
      streamMess.setView(result);
    }
    if (articleTypeName.equals(UpdateArticleMess.UpdateArticleType.LIKES.name())) { // 点赞
      streamMess.setLike(result);
    }
    if (articleTypeName.equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())) { // 评论
      streamMess.setComment(result);
    }
    if (articleTypeName.equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())) { // 收藏
      streamMess.setCollect(result);
    }

    return streamMess;
  }
}
